package _01_quick_start._8_2_5_back_pressure_in_action

import akka.stream._
import akka.stream.scaladsl._
import akka.{NotUsed, Done}
import akka.actor.ActorSystem
import akka.util.ByteString
import scala.concurrent._
import scala.concurrent.duration._
import java.nio.file.Paths
import scala.concurrent.duration._

object Main extends App {

  implicit val system       = ActorSystem("QuickStart")
  implicit val materializer = ActorMaterializer()
  implicit val ec           = system.dispatcher

  final case class Author(handle: String)

  final case class Hashtag(name: String)

  final case class Tweet(author: Author, timestamp: Long, body: String) {
    def hashtags: Set[Hashtag] =
      body.split(" ").collect { case t if t.startsWith("#") => Hashtag(t) }.toSet
  }

  val akkaTag = Hashtag("#akka")

  /*
   * 非结构化`streaming`处理
   */

  // source
  val tweets: Source[Tweet, NotUsed] = Source(
    Set(
      Tweet(Author("Martin"), System.currentTimeMillis(), "Ordersky #akka creator"),
      Tweet(Author("Barudisu"), System.currentTimeMillis(), "#akka leaner"),
      Tweet(Author("Petter"), System.currentTimeMillis(), "Mahina #scala practise")
    ))

  val done = tweets
             .buffer(10, OverflowStrategy.dropHead)   // Back-pressure 在flow添加一个固定大小元素的upstream,这里为10
             .map(_.author)                           // 取决于`akka.stream.OverflowStrategy`，当下游出现反作用力(back-pressure)
             .runWith(Sink foreach println)           // 该buffer将drop*(DropHead,DropTail,DropBuffer,never backpressures)

  done.onComplete(_ => system.terminate())
}
